package org.renhj.discovery;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.renhj.config.ZookeeperConfig;

import java.util.List;
import java.util.Objects;

public class ZookeeperDiscovery extends Discovery {


    private final static String ZOOKEEPER_HOST = ZookeeperConfig.ZOOKEEPER_HOST;
    private final static int ZOOKEEPER_PORT = ZookeeperConfig.ZOOKEEPER_PORT;
    private final static String PROVIDER_PATH = ZookeeperConfig.PROVIDER_PATH;
    private final static String CONSUMER_PATH = ZookeeperConfig.CONSUMER_PATH;
    private final static String PATH_SEPARATOR = ZookeeperConfig.PATH_SEPARATOR;


    private static final ZookeeperDiscovery zookeeperDiscovery = null;
    private final CuratorFramework curatorFramework;

    private List<String> providers;


    private ZookeeperDiscovery(String host, int port) {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(5000, 3);
        curatorFramework = CuratorFrameworkFactory.builder()
                .retryPolicy(retryPolicy)
                .connectionTimeoutMs(5000)
                .connectString(host + ":" + port)
                .namespace("curator")
                .build();
        curatorFramework.start();
    }

    public static ZookeeperDiscovery getInstance() {
        synchronized (ZookeeperDiscovery.class) {
            return Objects.requireNonNullElseGet(zookeeperDiscovery, () -> new ZookeeperDiscovery(ZOOKEEPER_HOST, ZOOKEEPER_PORT));
        }
    }

    // 服务发现
    @Override
    public List<String> discovery(String service) throws Exception {
        providers = curatorFramework.getChildren().forPath(pathJoin(PROVIDER_PATH, service));
        return providers;
    }

    // 注册监听
    public void registerListener(String service, PathChildrenCacheListener pathChildrenCacheListener) throws Exception {
        String path = pathJoin(PROVIDER_PATH, service);
        PathChildrenCache childrenCache = new PathChildrenCache(curatorFramework, path, true);
        childrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
        childrenCache.getListenable().addListener(pathChildrenCacheListener);
    }

}
